package cn._51doit.day10.asyncIOSQL;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.concurrent.TimeUnit;

/**
 * @create: 2021-10-27 20:04
 * @author: 今晚打脑斧先森
 * @program: AsyncQueryFromMySQL
 * @Description:
 *      将获取到的数据，写到sql中
 **/
public class AsyncQueryFromMySQL {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);
        int capacity = 20;
        DataStream<Tuple2<String, String>> result = AsyncDataStream.orderedWait(
                lines, //输入的数据流
                new MySQLAsyncFunction(capacity), //异步查询的Function实例
                3000, //超时时间
                TimeUnit.MILLISECONDS, //时间单位
                capacity); //最大异步并发请求数量
        result.print();
        env.execute();

    }

}
